package com.atguigu.dataStreamApi.source;


import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author WEIYUNHUI
 * @date 2024/11/15 8:54
 *
 * FileSystem 连接器
 *
 *   FileSource
 *
 *
 * Source API:   fromSource()
 *
 * SourceFunction API:  addSource()
 */

public class Flink02_FileSource {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //streamFormat:连接器，从文本中读数据
        //创建文本连接器对象
        FileSource<String> fileSource = FileSource.forRecordStreamFormat(
                        new TextLineInputFormat(),//输入文件格式是文本，按行输入
                        new Path("F:\\代码\\11_12flink\\FlinkTutorial\\input\\word.txt")//文件路径，绝对路径
                )
                .build();

        DataStreamSource<String> ds
                = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "filesource");


        ds.print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}